package com.longyb.mylive.server.jt1078;

import com.longyb.mylive.client.HandShakeEncoder;
import com.longyb.mylive.client.RtmpClientHandler;
import com.longyb.mylive.server.handlers.ChunkDecoder;
import com.longyb.mylive.server.handlers.ChunkEncoder;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioChannelOption;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * TODO
 *
 * @author chengxuwei
 * @date 2024-04-28
 * @since 1.0.0
 */
@Slf4j
public class JT1078Client {
  private static final EventLoopGroup GLOBAL_WORK_GROUP = new NioEventLoopGroup(10);
  private static final EventLoopGroup GLOBAL_HAND_GROUP = new NioEventLoopGroup(10);
  private Channel channel;
  private int connRetry = 3;


  /**
   * 建立一个连接，共享Eventloop
   *
   * @param host
   * @param port
   */
  /**
   * 打开连接
   * @param host
   * @param port
   */
  public void connect(String host, int port) {

    // 配置客户端NIO线程组
    //首先创建客户端处理I/O读写的NioEventLoop Group线程组
    try {

      //继续创建客户端辅助启动类Bootstrap，随后需要对其进行配置。
      //与服务端不同的是，它的Channel需要设置为NioSocketChannel
      //然后为其添加handler。
      // 在初始化它的时候将它的ChannelHandler设置到ChannelPipeline中，用于处理网络I/O事件。
      Bootstrap bootstrap = new Bootstrap();


      bootstrap.group(GLOBAL_WORK_GROUP);
      bootstrap.channel(NioSocketChannel.class);
      bootstrap.option(NioChannelOption.TCP_NODELAY, true);
      bootstrap.option(NioChannelOption.SO_REUSEADDR, true);
      bootstrap.option(NioChannelOption.SO_KEEPALIVE, false);
      bootstrap.option(NioChannelOption.SO_RCVBUF, 2048);
      bootstrap.option(NioChannelOption.SO_SNDBUF, 2048);
      //增加编解码工具
      bootstrap.handler(new ChannelInitializer<SocketChannel>() {
        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
          super.channelInactive(ctx);

        }

        @Override
        public void initChannel(SocketChannel ch) {
          ch.pipeline().addLast("idleStateHandler", new IdleStateHandler(10, 10, 10, TimeUnit.MINUTES));
          /**
           * maxFrameLength最大包长
           * lengthFieldOffset 长度字段偏移，第一个字段不是长度字段
           * lengthFieldLength 长度字段字节数
           * lengthAdjustment 长度字段修正，针对有特定头的
           * initialBytesToStrip 丢弃字节
           * failFast 快速失败
           *
           * 坑，默认lengthAdjustment不算长度字段
           */
          ch.pipeline().addLast("logging", new LoggingHandler(LogLevel.INFO));

          //跳分隔符，快速失败

          ch.pipeline()
            .addLast(new HandShakeEncoder())
            .addLast(new ChunkDecoder())
            .addLast(new ChunkEncoder())
            .addLast(GLOBAL_HAND_GROUP,new RtmpClientHandler());


        }
      });
      //然后调用同步方法等待连接成功。
      ChannelFuture channelFuture = bootstrap.connect(host, port).sync();

      /**
       * 监听
       */
      channelFuture.addListener(new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture channelFuture) throws Exception {

          //任务完成
          if (channelFuture.isDone()) {
            if (channelFuture.isSuccess()) {
              // 成功
              log.info("客户端连接成功:{},{}", host, port);
              channel = channelFuture.channel();
            } else if (channelFuture.isCancelled()) {
              // 被取消
              log.info("客户端连接取消:{},{}", host, port);
            } else {
              //重连交给后端线程执行
              if ((connRetry--) > 0) {
                channelFuture.channel().eventLoop().schedule(() -> {
                  log.error("重连服务端...");
                  connect(host, port);
                }, 3000, TimeUnit.MILLISECONDS);
              }
              // 异常
              log.error("客户端连接异常,3秒后重试:{},{},{}", host, port, connRetry, channelFuture.cause());
            }
          }
        }
      });
      channel = channelFuture.channel();
    } catch (InterruptedException e) {
      log.error("连接中断", e);
    }

  }


  public void setCallback() {

  }


  public boolean isConnect() {
    if (null == channel) {
      return false;
    }
    return channel.isActive();
  }

  /**
   * @param buf
   */
  public void write(Object buf) {
    if (null == channel) {
      log.warn("channel is null or close");
    } else {
      channel.writeAndFlush(buf);
    }
  }
  public void close(){
    if(channel !=null && channel.isActive()){
      channel.close();
    }
  }

  /**
   * 握手处理
   */
  public void handshake(){
    //1. 发送c0,c1

    //2. 接收s0,s1,s2->回复 c2



  }

  public static void main(String[] args) throws Exception {
    com.longyb.mylive.client.RtmpClient client=new com.longyb.mylive.client.RtmpClient();
    /**
     * 推送monibuca plugin
     */
    client.connect("127.0.0.1",7222);

    List<JT1078Msg> list = ReadJT1078Bin.readFile();
    while (true){
      for (int i = 0; i < list.size(); i++) {
        JT1078Msg msg = list.get(i);
        Thread.sleep(msg.getLastFrameMs());
        client.write(msg.getOrigin());
      }

    }

  }
}